package com.flink.hbase.statemanagement.managed;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * MapState托管状态示例
 * 
 * 展示如何使用MapState进行状态管理：
 * 1. 商品库存管理
 * 2. 用户设备统计
 * 3. 页面访问计数器
 * 4. 多维度指标统计
 * 
 * @author Flink Team
 */
public class MapStateExample {
    
    /**
     * 库存变动事件
     */
    public static class InventoryEvent {
        public String warehouseId;
        public String productId;
        public String operation; // ADD, REMOVE, UPDATE
        public int quantity;
        public long timestamp;
        
        public InventoryEvent() {}
        
        public InventoryEvent(String warehouseId, String productId, String operation, int quantity, long timestamp) {
            this.warehouseId = warehouseId;
            this.productId = productId;
            this.operation = operation;
            this.quantity = quantity;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("InventoryEvent{warehouseId='%s', productId='%s', operation='%s', quantity=%d, timestamp=%d}", 
                warehouseId, productId, operation, quantity, timestamp);
        }
    }
    
    /**
     * 用户设备事件
     */
    public static class UserDeviceEvent {
        public String userId;
        public String deviceId;
        public String deviceType;
        public String osVersion;
        public long timestamp;
        
        public UserDeviceEvent() {}
        
        public UserDeviceEvent(String userId, String deviceId, String deviceType, String osVersion, long timestamp) {
            this.userId = userId;
            this.deviceId = deviceId;
            this.deviceType = deviceType;
            this.osVersion = osVersion;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("UserDeviceEvent{userId='%s', deviceId='%s', deviceType='%s', osVersion='%s', timestamp=%d}", 
                userId, deviceId, deviceType, osVersion, timestamp);
        }
    }
    
    /**
     * 页面访问事件
     */
    public static class PageViewEvent {
        public String userId;
        public String pageId;
        public String sessionId;
        public String referrer;
        public long timestamp;
        
        public PageViewEvent() {}
        
        public PageViewEvent(String userId, String pageId, String sessionId, String referrer, long timestamp) {
            this.userId = userId;
            this.pageId = pageId;
            this.sessionId = sessionId;
            this.referrer = referrer;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("PageViewEvent{userId='%s', pageId='%s', sessionId='%s', referrer='%s', timestamp=%d}", 
                userId, pageId, sessionId, referrer, timestamp);
        }
    }
    
    /**
     * 多维度指标事件
     */
    public static class MetricEvent {
        public String service;
        public String metric;
        public String dimension;
        public double value;
        public long timestamp;
        
        public MetricEvent() {}
        
        public MetricEvent(String service, String metric, String dimension, double value, long timestamp) {
            this.service = service;
            this.metric = metric;
            this.dimension = dimension;
            this.value = value;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("MetricEvent{service='%s', metric='%s', dimension='%s', value=%.2f, timestamp=%d}", 
                service, metric, dimension, value, timestamp);
        }
    }
    
    /**
     * 商品库存管理函数
     * 使用MapState管理各个商品的库存数量
     */
    public static class InventoryManager extends RichFlatMapFunction<InventoryEvent, Tuple3<String, String, Integer>> {
        
        // 商品库存状态：productId -> quantity
        private MapState<String, Integer> inventoryState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建MapState描述符
            MapStateDescriptor<String, Integer> inventoryDescriptor = new MapStateDescriptor<>(
                "inventory-state", // 状态名称
                String.class, // Key类型
                Integer.class // Value类型
            );
            
            // 获取状态
            inventoryState = getRuntimeContext().getMapState(inventoryDescriptor);
        }
        
        @Override
        public void flatMap(InventoryEvent event, Collector<Tuple3<String, String, Integer>> collector) throws Exception {
            // 获取当前库存
            Integer currentStock = inventoryState.get(event.productId);
            if (currentStock == null) {
                currentStock = 0;
            }
            
            // 根据操作类型更新库存
            int newStock = currentStock;
            switch (event.operation) {
                case "ADD":
                    newStock = currentStock + event.quantity;
                    break;
                case "REMOVE":
                    newStock = Math.max(0, currentStock - event.quantity);
                    break;
                case "UPDATE":
                    newStock = event.quantity;
                    break;
            }
            
            // 更新状态
            inventoryState.put(event.productId, newStock);
            
            // 输出结果
            collector.collect(new Tuple3<>(event.warehouseId, event.productId, newStock));
            
            // 库存预警
            if (newStock < 10) {
                System.out.println("⚠️ 低库存预警: " + event.productId + " 库存仅剩 " + newStock);
            }
            
            System.out.println(String.format("📦 仓库 %s 商品 %s 库存变更: %d -> %d (操作: %s)", 
                event.warehouseId, event.productId, currentStock, newStock, event.operation));
        }
    }
    
    /**
     * 用户设备统计函数
     * 使用MapState统计用户使用的不同设备类型
     */
    public static class UserDeviceCounter extends KeyedProcessFunction<String, UserDeviceEvent, Tuple3<String, String, Integer>> {
        
        // 设备类型统计状态：deviceType -> count
        private MapState<String, Integer> deviceCountState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<String, Integer> deviceCountDescriptor = new MapStateDescriptor<>(
                "device-count-state", String.class, Integer.class);
            
            deviceCountState = getRuntimeContext().getMapState(deviceCountDescriptor);
        }
        
        @Override
        public void processElement(UserDeviceEvent event, Context context, 
                                 Collector<Tuple3<String, String, Integer>> collector) throws Exception {
            
            // 获取当前设备类型计数
            Integer currentCount = deviceCountState.get(event.deviceType);
            if (currentCount == null) {
                currentCount = 0;
            }
            
            // 更新计数
            int newCount = currentCount + 1;
            deviceCountState.put(event.deviceType, newCount);
            
            // 输出结果
            collector.collect(new Tuple3<>(event.userId, event.deviceType, newCount));
            
            // 输出用户设备使用统计
            Map<String, Integer> allDeviceCounts = new HashMap<>();
            for (Map.Entry<String, Integer> entry : deviceCountState.entries()) {
                allDeviceCounts.put(entry.getKey(), entry.getValue());
            }
            
            System.out.println(String.format("📱 用户 %s 设备统计: %s", event.userId, allDeviceCounts));
        }
    }
    
    /**
     * 页面访问计数器函数
     * 使用MapState统计每个页面的访问次数
     */
    public static class PageViewCounter extends KeyedProcessFunction<String, PageViewEvent, Tuple3<String, String, Integer>> {
        
        // 页面访问计数状态：pageId -> count
        private MapState<String, Integer> pageViewCountState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<String, Integer> pageViewCountDescriptor = new MapStateDescriptor<>(
                "page-view-count-state", String.class, Integer.class);
            
            pageViewCountState = getRuntimeContext().getMapState(pageViewCountDescriptor);
        }
        
        @Override
        public void processElement(PageViewEvent event, Context context, 
                                 Collector<Tuple3<String, String, Integer>> collector) throws Exception {
            
            // 获取当前页面访问计数
            Integer currentCount = pageViewCountState.get(event.pageId);
            if (currentCount == null) {
                currentCount = 0;
            }
            
            // 更新计数
            int newCount = currentCount + 1;
            pageViewCountState.put(event.pageId, newCount);
            
            // 输出结果
            collector.collect(new Tuple3<>(event.userId, event.pageId, newCount));
            
            System.out.println(String.format("🔍 用户 %s 访问页面 %s: 第 %d 次", 
                event.userId, event.pageId, newCount));
            
            // 每访问10次输出一次统计
            if (newCount % 10 == 0) {
                System.out.println("📊 用户 " + event.userId + " 页面访问统计达到 " + newCount + " 次");
            }
        }
    }
    
    /**
     * 多维度指标统计函数
     * 使用MapState实现多维度指标聚合
     */
    public static class MultiDimensionMetricAggregator extends KeyedProcessFunction<String, MetricEvent, Tuple3<String, String, Double>> {
        
        // 多维度指标状态：metric_dimension -> aggregated_value
        private MapState<String, Double> metricAggregationState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<String, Double> metricAggregationDescriptor = new MapStateDescriptor<>(
                "metric-aggregation-state", String.class, Double.class);
            
            metricAggregationState = getRuntimeContext().getMapState(metricAggregationDescriptor);
        }
        
        @Override
        public void processElement(MetricEvent event, Context context, 
                                 Collector<Tuple3<String, String, Double>> collector) throws Exception {
            
            // 构建维度键
            String dimensionKey = event.metric + "_" + event.dimension;
            
            // 获取当前聚合值
            Double currentValue = metricAggregationState.get(dimensionKey);
            if (currentValue == null) {
                currentValue = 0.0;
            }
            
            // 聚合新值（这里使用累加，可以根据需要改为其他聚合方式）
            double newValue = currentValue + event.value;
            metricAggregationState.put(dimensionKey, newValue);
            
            // 输出结果
            collector.collect(new Tuple3<>(event.service, dimensionKey, newValue));
            
            System.out.println(String.format("📈 服务 %s 指标聚合: %s = %.2f", 
                event.service, dimensionKey, newValue));
            
            // 输出所有维度的聚合结果
            if (newValue % 100 == 0) { // 每100的倍数输出一次完整统计
                System.out.println("📊 服务 " + event.service + " 完整指标统计:");
                for (Map.Entry<String, Double> entry : metricAggregationState.entries()) {
                    System.out.println("  " + entry.getKey() + ": " + entry.getValue());
                }
            }
        }
    }
    
    /**
     * 模拟库存变动事件源
     */
    public static class MockInventoryEventSource implements SourceFunction<InventoryEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] warehouseIds = {"warehouse1", "warehouse2", "warehouse3"};
        private final String[] productIds = {"product1", "product2", "product3", "product4", "product5"};
        private final String[] operations = {"ADD", "REMOVE", "UPDATE"};
        
        @Override
        public void run(SourceContext<InventoryEvent> ctx) throws Exception {
            while (running) {
                InventoryEvent event = new InventoryEvent(
                    warehouseIds[random.nextInt(warehouseIds.length)],
                    productIds[random.nextInt(productIds.length)],
                    operations[random.nextInt(operations.length)],
                    random.nextInt(50) + 1, // 1-50随机数量
                    System.currentTimeMillis()
                );
                
                ctx.collect(event);
                Thread.sleep(2000 + random.nextInt(3000)); // 2-5秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟用户设备事件源
     */
    public static class MockUserDeviceEventSource implements SourceFunction<UserDeviceEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] deviceTypes = {"mobile", "desktop", "tablet", "tv"};
        private final String[] osVersions = {"iOS15", "Android12", "Windows11", "macOS12"};
        private int deviceIdCounter = 1;
        
        @Override
        public void run(SourceContext<UserDeviceEvent> ctx) throws Exception {
            while (running) {
                UserDeviceEvent event = new UserDeviceEvent(
                    userIds[random.nextInt(userIds.length)],
                    "device_" + (deviceIdCounter++),
                    deviceTypes[random.nextInt(deviceTypes.length)],
                    osVersions[random.nextInt(osVersions.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(event);
                Thread.sleep(1500 + random.nextInt(2500)); // 1.5-4秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟页面访问事件源
     */
    public static class MockPageViewEventSource implements SourceFunction<PageViewEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] pageIds = {"home", "product", "cart", "checkout", "profile"};
        private final String[] referrers = {"google", "facebook", "direct", "email"};
        private int sessionIdCounter = 1;
        
        @Override
        public void run(SourceContext<PageViewEvent> ctx) throws Exception {
            while (running) {
                PageViewEvent event = new PageViewEvent(
                    userIds[random.nextInt(userIds.length)],
                    pageIds[random.nextInt(pageIds.length)],
                    "session_" + (sessionIdCounter++),
                    referrers[random.nextInt(referrers.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(event);
                Thread.sleep(1000 + random.nextInt(2000)); // 1-3秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟多维度指标事件源
     */
    public static class MockMetricEventSource implements SourceFunction<MetricEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] services = {"user-service", "order-service", "payment-service"};
        private final String[] metrics = {"response_time", "throughput", "error_rate"};
        private final String[] dimensions = {"region_us", "region_eu", "region_asia"};
        
        @Override
        public void run(SourceContext<MetricEvent> ctx) throws Exception {
            while (running) {
                MetricEvent event = new MetricEvent(
                    services[random.nextInt(services.length)],
                    metrics[random.nextInt(metrics.length)],
                    dimensions[random.nextInt(dimensions.length)],
                    random.nextDouble() * 100.0, // 0-100随机值
                    System.currentTimeMillis()
                );
                
                ctx.collect(event);
                Thread.sleep(3000 + random.nextInt(2000)); // 3-5秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 运行MapState示例
     */
    public static void runMapStateExample(StreamExecutionEnvironment env) throws Exception {
        System.out.println("🚀 开始运行MapState示例...");
        
        // 1. 商品库存管理示例
        DataStream<InventoryEvent> inventoryEventStream = env.addSource(new MockInventoryEventSource())
            .name("库存变动事件源");
        
        DataStream<Tuple3<String, String, Integer>> inventoryManagementStream = inventoryEventStream
            .keyBy(event -> event.warehouseId)
            .flatMap(new InventoryManager())
            .name("商品库存管理");
        
        inventoryManagementStream.print("库存管理");
        
        // 2. 用户设备统计示例
        DataStream<UserDeviceEvent> userDeviceEventStream = env.addSource(new MockUserDeviceEventSource())
            .name("用户设备事件源");
        
        DataStream<Tuple3<String, String, Integer>> deviceCountStream = userDeviceEventStream
            .keyBy(event -> event.userId)
            .process(new UserDeviceCounter())
            .name("用户设备统计");
        
        deviceCountStream.print("设备统计");
        
        // 3. 页面访问计数器示例
        DataStream<PageViewEvent> pageViewEventStream = env.addSource(new MockPageViewEventSource())
            .name("页面访问事件源");
        
        DataStream<Tuple3<String, String, Integer>> pageViewCountStream = pageViewEventStream
            .keyBy(event -> event.userId)
            .process(new PageViewCounter())
            .name("页面访问计数器");
        
        pageViewCountStream.print("页面访问");
        
        // 4. 多维度指标统计示例
        DataStream<MetricEvent> metricEventStream = env.addSource(new MockMetricEventSource())
            .name("多维度指标事件源");
        
        DataStream<Tuple3<String, String, Double>> metricAggregationStream = metricEventStream
            .keyBy(event -> event.service)
            .process(new MultiDimensionMetricAggregator())
            .name("多维度指标统计");
        
        metricAggregationStream.print("指标统计");
        
        System.out.println("✅ MapState示例配置完成");
    }
    
    /**
     * 单独运行MapState示例的主方法
     */
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 配置状态后端
        com.flink.hbase.statemanagement.config.StateBackendConfig
            .configureDevRocksDB(env);
        
        // 运行示例
        runMapStateExample(env);
        
        // 执行作业
        env.execute("MapState托管状态示例");
    }
} 